package mspbots.hubspot.client.core.companies;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import mspbots.hubspot.client.core.AbstractToolsUtil;
import mspbots.hubspot.client.core.BaseApiService;
import mspbots.hubspot.common.*;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.UUID;

/**
 * data.connector.hubspot.core.contacts.ContactService
 *
 * @author Alex bob(https://github.com/vnobo)
 * @date Created by 2020/3/31
 */
@Log4j2
@Service
@RequiredArgsConstructor
public class CompaniesServiceImpl extends AbstractToolsUtil implements BaseApiService {

    @Override
    public Flux<JsonNode> properties(Long tenantId) {
        return this.hubSpotClient.get("/crm/v3/properties/tickets", null)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId))
                .flatMapMany(jsonNode -> Flux.fromIterable(jsonNode.withArray("results")));
    }

    @Override
    public Flux<JsonNode> search(Long tenantId, QueryBody queryBody) {
        return this.hubSpotClient.fetch("/crm/v3/objects/companies/search", queryBody)
                .subscriberContext(ctx -> ctx.put("tenantId", tenantId));
    }

    @Override
    public Mono<Void> syncProperties(Long tenantId) {
        ResultSync resultSync = ResultSync.withDefault("COMPANIES-ENTITY-PROPERTIES");
        return this.properties(tenantId)
                .subscriberContext(ctx -> ctx.put("resultSync", resultSync))
                .doOnNext(result -> this.syncPropertiesResult(tenantId,
                        resultSync.getRequestId(), result))
                .then();
    }


    private void syncPropertiesResult(Long tenantId, UUID id, JsonNode resultNode) {
        this.sendEntity(RabbitKeys.PROPERTIES_SYNC_QUEUE_KEY, ResultMsg.builder()
                .requestId(id).action("Companies")
                .entity(resultNode).tenantId(tenantId)
                .syncDate(LocalDateTime.now()).build());
    }

    @Override
    public Mono<Void> syncData(Long tenantId) {
        ResultSync resultSync = ResultSync.withDefault("COMPANIES-SYNC");
        return this.dateCacheFilterContacts(tenantId)
                .subscriberContext(ctx -> ctx.put("resultSync", resultSync))
                .doOnNext(result -> this.syncResult(tenantId, resultSync.getRequestId(), result))
                .then();
    }

    private void syncResult(Long tenantId, UUID id, JsonNode resultNode) {
        Long entityId = Long.parseLong(resultNode.get("id").textValue());
        log.debug("Companies result id {}", resultNode.get("id").textValue());
        this.sendEntity(RabbitKeys.COMPANIES_SYNC_QUEUE_KEY, ResultMsg.builder()
                .requestId(id).id(entityId)
                .action("none").entity(resultNode).tenantId(tenantId)
                .syncDate(LocalDateTime.now()).build());
    }

    private Flux<JsonNode> dateCacheFilterContacts(Long tenantId) {
        String redisAnchorKey = RedisKeys.COMPANIES_ANCHOR_REDIS_KEY_PREFIX + tenantId;
        return super.readAnchorTime(redisAnchorKey)
                .flatMapMany(syncTime -> this.search(tenantId,
                        QueryBody.withFilter(FilterExpression.builder()
                                .operator("GT").propertyName("hs_lastmodifieddate")
                                .value(syncTime.toString()).build())))
                .doOnNext(user -> super.recordAnchorTime(redisAnchorKey,
                        LocalDateTime.parse(user.get("updatedAt").textValue(), DateTimeFormatter.ISO_DATE_TIME)
                                .toInstant(ZoneOffset.UTC).toEpochMilli()));
    }

}
